#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | date_add(-2) | cst_ds }}'
name='中心停留时长汇总表'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "日期\t代理区\t中心名称\t中心编码\t操作总票数\t发件票数(中心-网点/集散)\t发件平均停留时长(h)\t发件停留时间36-48小时(票)\t发件停留时间超48小时(票)\t发车票数(中心-中心)\t发车平均停留时长(h)\t发车停留时间36-48小时(票)\t发车停留时间超48小时(票)\t有到无发票数或有发无到票数(票)\t超36小时占比\t超36-48小时占比\t超48小时占比\t">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 8G --num-executors 10 --executor-memory 8g --executor-cores 6  --conf spark.executor.memoryOverhead=4g --conf spark.driver.maxResultSize=3g -S -e "
  with base1 as (
select dt,
    arrival_site,
    arrival_site_code,
    waybill_no ,
    max(case when diff_time>0  and send_next_type=2 then 1 else 0 end) as last_time,
    max(case when diff_time>0   then 1 else 0 end) as youxiao
from (
    select
    to_date(t.arrival_time) as dt,
    arrival_site,
    arrival_site_code,
    waybill_no,
    send_next_type,
    case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
             else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 end diff_time
    from   jms_dm.dm_center_stop_time_dt t where  t.dt='{{ execution_date | date_add(-2) | cst_ds }}'
) t
group by    dt,
    arrival_site,
    arrival_site_code,
    waybill_no
),
base2 as (
select
    dt,
    arrival_site,
    arrival_site_code,
    waybill_no,
    max(fache) as fache,
    max(fanjian) as fanjian
from (
select
to_date(t.arrival_time) as dt,
arrival_site,
arrival_site_code,
waybill_no,
case when (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time)))  >0 and send_next_type=2  then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 else 0 end  as fache,
case when (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time)))  >0 and send_next_type<>2  then (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 else 0 end  as fanjian
from   jms_dm.dm_center_stop_time_dt t where t.send_time <> '0000-0000-0000' and t.arrival_time < t.send_time and t.dt='{{ execution_date | date_add(-2) | cst_ds }}'
)
group by     dt,
    arrival_site,
    arrival_site_code,
    waybill_no
)

select dt,
    t1.agent_name,
    arrival_site,
    arrival_site_code,
    sum(a),--操作总票数
    sum(c),--发件票数
    sum(aa1),--发件平均时长
    sum(g),--发件36
    sum(h),--发件48
    sum(d),--发车票数
    sum(aa),--发车平均时长
    sum(e),--发车36
    sum(f),--发车48
    sum(a)-sum(b) as aaaa,
    (sum(g)+sum(h)+sum(e)+sum(f) +(sum(a)-sum(b)) )/(sum(a)*1.00) as r1,
    (sum(g)+sum(e) )/(sum(a)*1.00) as r2,
    (sum(h)+sum(f) )/(sum(a)*1.00) as r3
from (
select
    dt ,
    arrival_site,
    arrival_site_code,
    count(waybill_no) as a ,
    sum(youxiao ) as b,
    sum(youxiao ) - sum(last_time ) as c,
    sum(last_time) as d,
    0 as e,
    0 as f,
    0 as g,
    0 as h,
    0 as e1,
    0 as f1,
    0 as g1,
    0 as h1,
    0 as aa,
    0 as aa1
from
    base1 t
group by     dt,
    arrival_site,
    arrival_site_code
union all
select
    dt,
    arrival_site,
    arrival_site_code,
    0 as a ,
    0 as b,
    0 as c,
    0 as d,
    sum(case when fache >=36 and fache<48 then 1 else 0 end ) as e,
    sum(case when fache >=48  then 1 else 0 end ) as f,
    sum(case when fanjian >=36 and fanjian<48 then 1 else 0 end ) as g,
    sum(case when fanjian >=48 then 1 else 0 end ) as h,
    0 as e1,
    0 as f1,
    0 as g1,
    0 as h1,
    0 as aa,
    0 as aa1
from
    base2 t
group by     dt,
    arrival_site,
    arrival_site_code
union all
select
    dt,
    arrival_site,
    arrival_site_code,
    0 as a ,
    0 as b,
    0 as c,
    0 as d,
    0 as e,
    0 as f,
    0 as g,
    0 as h,
    sum(case when m1 >=36 and m1<48 then 1 else 0 end ) as e1,
    sum(case when m1 >=48  then 1 else 0 end ) as f1,
    sum(case when m2 >=36 and m2<48 then 1 else 0 end ) as g1,
    sum(case when m2 >=48 then 1 else 0 end ) as h1,
    avg(m1) as aa,
    avg(m2) as aa1
from (
select
    to_date(t.arrival_time) as dt,
    arrival_site,
    arrival_site_code,
    0 as a ,
    0 as b,
    0 as c,
    0 as d,
    0 as e,
    0 as f,
    0 as g,
    0 as h,
    case when (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) >0 and send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 else 0 end  as m1,
    case when (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) >0 and send_next_type<>2  then (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 else 0 end  as m2
from   jms_dm.dm_center_stop_time_dt t where t.send_time <> '0000-0000-0000' and t.arrival_time < t.send_time  and t.dt='{{ execution_date | date_add(-2) | cst_ds }}'
)
group by     dt,
    arrival_site,
    arrival_site_code
) t
left join  jms_dim.dim_network_whole_massage t1 on t.arrival_site_code=t1.code
    group by     dt,
    arrival_site,
    arrival_site_code,t1.agent_name
;
 " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt